package com.atguigu.flink.chapter05.sink;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.sql.PreparedStatement;
import java.sql.SQLException;

public class JdbcSinkDemo {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        SingleOutputStreamOperator<Tuple2<String, Long>> resultStream = env
                .socketTextStream("hadoop162", 8888)
                .flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public void flatMap(String value, Collector<Tuple2<String, Long>> out) throws Exception {
                        for (String word : value.split(" ")) {
                            out.collect(Tuple2.of(word, 1L));
                        }
                    }
                })
                .keyBy(t -> t.f0)
                .sum(1);

        // hello  3
        resultStream.addSink(JdbcSink.sink(
//            "insert into wc(word, ct)values(?, ?)",
                "replace into wc(word, ct)values(?, ?)",  // 当主键不存在就 insert,如果主键存在就 update
                new JdbcStatementBuilder<Tuple2<String, Long>>() {
                    @Override
                    public void accept(PreparedStatement ps,
                                       Tuple2<String, Long> ts) throws SQLException {
                        ps.setString(1, ts.f0);
                        ps.setLong(2, ts.f1);

                    }
                },
                new JdbcExecutionOptions.Builder()
                        .withBatchSize(10)
                        .withBatchIntervalMs(2000)
                        .withMaxRetries(5)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName("com.mysql.cj.jdbc.Driver") //
                        .withUrl("jdbc:mysql://hadoop162:3306/flink?useSSL=false")
                        .withUsername("root")
                        .withPassword("aaaaaa")
                        .build()
        ));
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}


//public class JdbcSinkDemo {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port", 2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//        SingleOutputStreamOperator<Tuple2<String, Long>> resultStream = env
//                .socketTextStream("hadoop162", 8888)
//                .flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
//                    @Override
//                    public void flatMap(String value, Collector<Tuple2<String, Long>> out) throws Exception {
//                        for (String word : value.split(" ")) {
//                            out.collect(Tuple2.of(word, 1L));
//                        }
//                    }
//                })
//                .keyBy(t -> t.f0)
//                .sum(1);
//
//        resultStream.addSink(JdbcSink.sink(
//                "replace into wc(word,ct) values(?,?)",
//                new JdbcStatementBuilder<Tuple2<String, Long>>() {
//                    @Override
//                    public void accept(PreparedStatement ps,
//                                       Tuple2<String, Long> ts) throws SQLException {
//                        ps.setString(1,ts.f0);
//                        ps.setLong(2,ts.f1);
//                    }
//                },
//                new JdbcExecutionOptions.Builder()
//                        .withBatchSize(10)
//                        .withBatchIntervalMs(2000)
//                        .withMaxRetries(5)
//                        .build(),
//                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
//                        .withDriverName("com.mysql.cj.jdbc.Driver")
//                        .withUrl("jdbc:mysql://hadoop102:3306/flink?useSSL=false")
//                        .withUsername("root")
//                        .withPassword("000000")
//                        .build()
//        ));
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}


//public class JdbcSinkDemo {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port", 2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//        SingleOutputStreamOperator<Tuple2<String, Long>> resultStream = env
//                .socketTextStream("hadoop162", 8888)
//                .flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
//                    @Override
//                    public void flatMap(String value, Collector<Tuple2<String, Long>> out) throws Exception {
//                        for (String word : value.split(" ")) {
//                            out.collect(Tuple2.of(word, 1L));
//                        }
//                    }
//                })
//                .keyBy(t -> t.f0)
//                .sum(1);
//
//        resultStream.addSink(JdbcSink.sink(
//                "replace into wc(word,ct) values(?,?)",
//                new JdbcStatementBuilder<Tuple2<String, Long>>() {
//                    @Override
//                    public void accept(PreparedStatement ps,
//                                       Tuple2<String, Long> ts) throws SQLException {
//                        ps.setString(1,ts.f0);
//                        ps.setLong(2,ts.f1);
//                    }
//                },
//                new JdbcExecutionOptions.Builder()
//                        .withBatchSize(10)
//                        .withBatchIntervalMs(2000)
//                        .withMaxRetries(5)
//                        .build(),
//
//                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
//                        .withDriverName("com.mysql.cj.jdbc.Driver")
//                        .withUrl("jdbc:mysql://hadoop102/flink?useSSL=false")
//                        .withUsername("root")
//                        .withPassword("000000")
//                        .build()
//        ));
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}
